package com.lxj.app.dwd;

import com.lxj.utils.MyKafkaUtil;
import com.lxj.utils.MysqlUtil;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
//9.4 交易域加购事务事实表
/**
 * 交易域加购事实表应用
 * 负责处理用户添加购物车的交易数据，将原始日志转化为结构化的加购事实表
 * 主要功能点：
 * - 从Kafka读取基础日志数据
 * - 过滤出加购相关的交互事件
 * - 关联商品和用户维度信息
 * - 提取加购时间、数量、价格等交易信息
 * - 将处理后的数据写入Kafka主题
 * 
 * 数据流向：
 * 基础日志数据 -> DwdTradeCartAdd -> 处理后写入dwd_trade_cart_add主题
 */
public class DwdTradeCartAdd {

    /**
     * 主函数，用于构建并执行加购事务事实表的流处理任务。
     *
     * @param args 命令行参数（未使用）
     * @throws Exception 任务执行过程中可能出现的异常
     */
    public static void main(String[] args) throws Exception {
        // 创建Flink流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1（生产环境中应根据Kafka主题分区数设置）
        env.setParallelism(1);

        // 创建Table API运行时环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 启用CheckPoint机制，间隔5分钟，保证Exactly-Once语义
        env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        // 设置CheckPoint超时时间为10分钟
        env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        // 设置最大同时进行的CheckPoints数量为2
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        // 设置重启策略：最多重启3次，每次间隔5秒
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));
        // 设置状态后端为HashMapStateBackend（本地开发使用）
        env.setStateBackend(new HashMapStateBackend());
        // 设置CheckPoint存储路径为HDFS
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/2212a/ck");
        // 设置Hadoop用户名称
        System.setProperty("HADOOP_USER_NAME", "bw");

        // TODO 2. 创建 topic_db 表
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("dwd_trade_cart_add"));
        System.out.println("创建topic_db表成功");
        
        // TODO 3. 过滤出加购数据
        // 使用SQL查询从Kafka中过滤出符合加购条件的数据
        Table cartAddTable = tableEnv.sqlQuery("" +
                "select " +
                "    `data`['id'] id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['sku_id'] sku_id, " +
                "    `data`['cart_price'] cart_price, " +
                "    if(`type`='insert',`data`['sku_num'],cast(cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int) as string)) sku_num, " +
                "    `data`['sku_name'] sku_name, " +
                "    `data`['is_checked'] is_checked, " +
                "    `data`['create_time'] create_time, " +
                "    `data`['operate_time'] operate_time, " +
                "    `data`['is_ordered'] is_ordered, " +
                "    `data`['order_time'] order_time, " +
                "    `data`['source_type'] source_type, " +
                "    `data`['source_id'] source_id, " +
                "    pt " +
                "from topic_db " +
                "where `database` = 'gmall' " +
                "and `table` = 'cart_info' " +
                "and (`type` = 'insert' " +
                "or (`type` = 'update'  " +
                "    and  " +
                "    `old`['sku_num'] is not null  " +
                "    and  " +
                "    cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int)))");

        // 将过滤后的加购数据注册为临时视图cart_info_table
        tableEnv.createTemporaryView("cart_info_table", cartAddTable);
        // 将加购数据转换为DataStream并打印测试
        tableEnv.toAppendStream(cartAddTable, Row.class).print(">>>>>>");

        // 查询并打印cart_info_table中的所有数据
        tableEnv.sqlQuery("select  *  from cart_info_table ").execute().print();

        // TODO 4. 读取MySQL的base_dic表作为LookUp表
        // 执行获取MySQL维度表base_dic的DDL语句
        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        // TODO 5. 关联两张表
        // 将加购表与维度表base_dic关联，获取来源类型名称
        Table cartAddWithDicTable = tableEnv.sqlQuery("" +
                "select " +
                "    ci.id, " +
                "    ci.user_id, " +
                "    ci.sku_id, " +
                "    ci.cart_price, " +
                "    ci.sku_num, " +
                "    ci.sku_name, " +
                "    ci.is_checked, " +
                "    ci.create_time, " +
                "    ci.operate_time, " +
                "    ci.is_ordered, " +
                "    ci.order_time, " +
                "    ci.source_type source_type_id, " +
                "    dic.dic_name source_type_name, " +
                "    ci.source_id " +
                "from cart_info_table ci " +
                "join base_dic FOR SYSTEM_TIME AS OF ci.pt as dic " +
                "on ci.source_type = dic.dic_code");
        // 注册临时视图为cart_add_dic_table
        tableEnv.createTemporaryView("cart_add_dic_table", cartAddWithDicTable);

        // TODO 6. 使用DDL方式创建加购事实表
        // 定义dwd_cart_add表结构，并将其写入到Kafka中
        tableEnv.executeSql("" +
                "create table dwd_cart_add( " +
                "    `id` STRING, " +
                "    `user_id` STRING, " +
                "    `sku_id` STRING, " +
                "    `cart_price` STRING, " +
                "    `sku_num` STRING, " +
                "    `sku_name` STRING, " +
                "    `is_checked` STRING, " +
                "    `create_time` STRING, " +
                "    `operate_time` STRING, " +
                "    `is_ordered` STRING, " +
                "    `order_time` STRING, " +
                "    `source_type_id` STRING, " +
                "    `source_type_name` STRING, " +
                "    `source_id` STRING " +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_cart_add"));

        // TODO 7. 将数据写出
        // 将cart_add_dic_table中的数据插入到目标表dwd_cart_add中
        tableEnv.executeSql("insert into dwd_cart_add select * from cart_add_dic_table");

        // TODO 8. 启动任务
        // 提交任务并开始执行
        env.execute("DwdTradeCartAdd");
    }



}
